package com.sunday.common.mq.rabbit.study.base.e15_ConsumerPrefetch;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * https://www.rabbitmq.com/consumer-prefetch.html
 * 消费者预取
 * 使用者预取是通道预取机制的扩展。
 * AMQP 0-9-1 指定了 basic.qos 方法，以便可以限制通道上未确认消息的数量（或 连接）在使用时（也称为“预取计数”）。不幸 该通道不是理想的范围 - 因为单个 通道可能从多个队列、通道和 队列需要为每条消息相互协调 发送以确保它们不会超过限制。这很慢 单台计算机，跨集群使用时非常慢。
 * 此外，对于许多用途，指定更自然 适用于每个使用者的预取计数。
 * 因此，RabbitMQ 在 basic.qos 方法中重新定义了全局标志的含义：
 * <p>
 * global	    AMQP 0-9-1 中prefetch_count的含义	                    RabbitMQ中prefetch_count的含义
 * false	    在通道(channel)上的所有消费者(consumers)之间共享	            分别应用于通道(channel)上的每个新消费者(consumer)
 * true	        在连接(connection)上的所有消费者(consumers)之间共享	        在通道(channel)上的所有消费者(consumers)之间共享
 */
@Slf4j
public class ConsumerPrefetch {

    /**
     * 单一消费者
     * 以下 Java 中的基本示例最多会收到 5 个 一次未确认的消息：
     * prefetchCount 值 0 被视为无限，允许任意数量的未确认 消息。
     */
    @Test
    @SneakyThrows
    public void singleConsumer() {

        String QUEUE_NAME = "single_Consumer_queue";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete(QUEUE_NAME);

            /**
             * 分别应用于通道(channel)上的每个新消费者(consumer)
             * @param prefetchSize 内容的最大数量(以服务器将交付的字节数(octet)，如果无限制，则为0
             * @param prefetchCount 服务器的最大消息数 如果无限制，我们会送货吗
             * @param global true如果设置应用于 整个渠道，而不是每个消费者
             */
//            channel.basicQos(0,5, false);
//            channel.basicQos(5);
            channel.basicQos(0);

            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            for (int i = 0; i < 10; i++) {
                channel.basicPublish("", QUEUE_NAME, null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            log.info("================================================");

            channel.basicConsume(QUEUE_NAME, false,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [x] Received '{}'", message);
                        try {
                            TimeUnit.SECONDS.sleep(5);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(60);

        }
    }

    /**
     * 独立消费者
     * 此示例在同一通道上启动两个使用者，每个使用者 将独立收到最多 3 个未确认 一次消息：
     *
     * 分别应用于通道(channel)上的每个新消费者(consumer)
     */
    @Test
    @SneakyThrows
    public void independentConsumers() {

        String QUEUE_NAME_1 = "independent_consumers_queue_1";
        String QUEUE_NAME_2 = "independent_consumers_queue_2";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete(QUEUE_NAME_1);
            channel.queueDelete(QUEUE_NAME_2);

            /**
             * 分别应用于通道(channel)上的每个新消费者(consumer)
             */
            channel.basicQos(3);

            channel.queueDeclare(QUEUE_NAME_1, false, false, false, null);
            channel.queueDeclare(QUEUE_NAME_2, false, false, false, null);

            for (int i = 0; i < 5; i++) {
                channel.basicPublish("", QUEUE_NAME_1, null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
                channel.basicPublish("", QUEUE_NAME_2, null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            log.info("================================================");

            channel.basicConsume(QUEUE_NAME_1, false,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [one][x] Received '{}'", message);
                        try {
                            TimeUnit.SECONDS.sleep(5);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    },
                    consumerTag -> {
                    });

            channel.basicConsume(QUEUE_NAME_2, false,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [two][x] Received '{}'", message);
                        try {
                            TimeUnit.SECONDS.sleep(5);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(60);

        }
    }

    /**
     * 在 RabbitMQ 中，channel.basicQos(prefetchCount, global)方法用于设置消费者的预取计数（prefetch count）和全局标志（global flag）。以下是对两种不同配置的区别解释：
     * 一、参数说明
     * prefetchCount：预取计数，表示在未确认之前，消费者可以预取的最大消息数量。
     * global：全局标志，决定预取计数是应用于每个新消费者还是在通道上的所有消费者之间共享。
     * 二、channel.basicQos(3, false)配置
     * 针对每个新消费者
     * 当global设置为false时，预取计数仅应用于通道上的每个新消费者。
     * 这意味着每个消费者可以独立地预取最多prefetchCount条消息，而不会受到其他消费者的影响。
     * 例如，如果有多个消费者同时连接到同一个通道，每个消费者都可以根据自己的处理能力预取不同数量的消息，最多不超过 3 条。
     * 灵活性
     * 这种配置适用于不同消费者具有不同处理能力或需求的场景。
     * 每个消费者可以根据自身情况调整预取的消息数量，以避免消息堆积或过度负载。
     * 三、channel.basicQos(3, true)配置
     * 所有消费者共享
     * 当global设置为true时，预取计数在通道上的所有消费者之间共享。
     * 这意味着整个通道上的所有消费者总共可以预取最多prefetchCount条消息。
     * 例如，如果有三个消费者连接到通道，并且预取计数设置为 3，那么这三个消费者总共最多可以预取 3 条消息。
     * 一致性控制
     * 这种配置适用于需要对通道上的所有消费者进行统一流量控制的场景。
     * 可以确保整个通道上的消息分配更加均衡，避免某些消费者过度预取而导致其他消费者无法获取消息。
     * 四、适用场景举例
     * false的适用场景
     * 多个消费者处理不同类型的任务，每个任务的处理时间和负载不同。此时，每个消费者可以根据自己的任务特点调整预取消息数量，提高处理效率。
     * 例如，一个消费者处理图像识别任务，另一个消费者处理文本分析任务，它们的处理速度和资源需求不同，可以分别设置合适的预取计数。
     * true的适用场景
     * 多个消费者处理相同类型的任务，需要保证整体的负载均衡和消息分配公平性。
     * 例如，在一个分布式计算系统中，多个消费者同时处理相同的计算任务，通过设置全局预取计数，可以确保每个消费者都有机会获取消息，避免某些消费者过度忙碌而其他消费者空闲。
     */

    /**
     * 多个使用者共享限制
     * AMQP 0-9-1 规范没有解释如果出现以下情况会发生什么 使用不同的全局值多次调用 basic.qos。
     * RabbitMQ将其解释为含义 两个预取限制应独立于 彼此;消费者只有在两者都没有收到新消息时才会收到新消息 已达到未确认消息的限制。
     *
     * 这两个消费者将只有4个未被承认 它们之间的消息，
     * 每个消息最多 3 条 消费者。
     * 这将比上述示例慢，因为 在通道和 用于强制实施全局限制的队列。
     */
    @Test
    @SneakyThrows
    public void configurableDefaultPrefetch() {

        String QUEUE_NAME_1 = "configurable_default_prefetch_queue_1";
        String QUEUE_NAME_2 = "configurable_default_prefetch_queue_2";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDelete(QUEUE_NAME_1);
            channel.queueDelete(QUEUE_NAME_2);

            /**
             * global :
             *      false   分别应用于通道(channel)上的每个新消费者(consumer)
             *      true    在通道(channel)上的所有消费者(consumers)之间共享
             */
            channel.basicQos(3, false);
            channel.basicQos(4, true);

            channel.queueDeclare(QUEUE_NAME_1, false, false, false, null);
            channel.queueDeclare(QUEUE_NAME_2, false, false, false, null);

            for (int i = 0; i < 10; i++) {
                channel.basicPublish("", QUEUE_NAME_1, null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
                channel.basicPublish("", QUEUE_NAME_2, null, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            log.info("================================================");

            channel.basicConsume(QUEUE_NAME_1, false,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [one][x] Received '{}'", message);
                        try {
                            TimeUnit.SECONDS.sleep(5);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    },
                    consumerTag -> {
                    });

            channel.basicConsume(QUEUE_NAME_2, false,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" [two][x] Received '{}'", message);
                        try {
                            TimeUnit.SECONDS.sleep(5);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(120);

        }
    }

}
